#include "MiRpc.h"
#include <unordered_map>
#include <list>
#include <iostream>
#ifdef __cplusplus
extern "C" {
#endif

    typedef void (*rpc_request_func)(void* self, void* ex, void* user_ctx);

    typedef void (*rpc_session_callback)(void* self, void* user_ctx);
    struct memory_manager
    {
        MemoryManager mm;
        memory_manager(void* mem, uint64_t mem_size, uint32_t max): mm(mem, mem_size, max){}
    };
    struct rpc
    {
        RdmaRpc rdma_rpc;
        rpc(struct memory_manager* mm) : rdma_rpc(mm->mm) {}
    };
    
    struct msg_base
    {
        uint32_t msg_len; //msg_total_len
        uint32_t sync; //msg can be free?
        uint64_t sign; //signature
        uint32_t free;
        uint16_t rpc_type; //rpc request type
        uint16_t status_code; // rpc response status code
        
        uint32_t data_len;
        char data[0];
    };
    struct rpc_execution 
    {
        struct msg_base* req_msg;
        struct msg_base* resp_msg;

        rpc_request_func cb;
        void* cb_arg;
        int cb_error;
        uint32_t self_id;
        uint64_t trigger_req;
        uint64_t remote_addr;
        uint32_t large_msg_len;
        uint32_t finish_len;

        uint64_t time_stamp;
        void *wait_free_rpc;
        uint32_t wait_free_sess_id;

    };

    struct rpc* rpc_create(void* mem, uint64_t mem_size, uint8_t flag, void* user_ctx)
    {
        struct memory_manager* m = new struct memory_manager(mem, mem_size, 262144);
        std::cout << "mem = " << m->mm.memory() << " mem_size = "<< m->mm.memory_size() << " max_msg = " << m->mm.max_msg_size()<<std::endl;
        m->mm.Init();
        struct rpc* new_rpc = new struct rpc(m);
        new_rpc->rdma_rpc.Init(flag);
        return new_rpc;
    }

    void rpc_destroy(struct rpc* self)
    {
        self->rdma_rpc.Destroy();
        delete self;
    }

    
    uint32_t rpc_bind_local_uri(struct rpc* self, char* uri)
    {
        return self->rdma_rpc.BindSess(uri);
    }


    void rpc_accept_session(struct rpc* self)
    {
        self->rdma_rpc.AcceptSess();
    }

    
    uint32_t rpc_create_session(struct rpc* self, char* uri, rpc_session_callback cb, void* cb_arg)
    {
        
        return self->rdma_rpc.ConnectSess(uri,cb,cb_arg);
    }

    void rpc_close_session(struct rpc* self, int sess_id)
    {
        //TODO::
        return;
    }

    void rpc_register_on_request(struct rpc* self, uint8_t rpc_type, rpc_request_func request_handler)
    {
        //TODO::
    
        self->rdma_rpc.RpcServiceRegister(rpc_type, request_handler);
    }

    void* alloc_msg_header(struct rpc* self)
    {
        return self->rdma_rpc.AllocMsg();
    }

    int free_msg_header(struct rpc* self, void* header)
    {
        return self->rdma_rpc.FreeMsg(header);
    }

    void alloc_remote_large_buf(struct msg_base* msg)
    {
        msg->status_code = RPC_LARGE_MSG_NOTIFY2;
    }

    
    void* prepare_msg_header(struct rpc* self, struct msg_base* header, uint8_t rpc_type,
    uint64_t req_addr, uint32_t req_len, uint64_t resp_addr, uint32_t resp_len)
    {
        return self->rdma_rpc.PrepareMsg((struct RpcMsg*)header, rpc_type);
    }

    int enqueue_request(struct rpc* self, int sess, struct msg_base* header, rpc_request_func cb , void* cb_arg, int batching)
    {
        //TODO
        return self->rdma_rpc.RpcCliWriteReq(sess, (struct RpcMsg*)header, cb, cb_arg);
    }

    int run_poll_once(struct rpc* self)
    {
        if(self->rdma_rpc.flag_ & RPC_SERVER)
        {
            return self->rdma_rpc.RpcServerHandleRx() + self->rdma_rpc.RpcServerHandleTx();
        } else {
            self->rdma_rpc.RpcCliHandleTx(); 
            return self->rdma_rpc.RpcCliHandleRx();
        }

    }
    void* alloc_large(struct rpc* self)
    {
        return self->rdma_rpc.AllocLarge();
    }
    void free_large(struct rpc *self, void* buf)
    {
        self->rdma_rpc.FreeLarge(buf);
    }

    void* alloc_small_piece(struct rpc* self)
    {
        return self->rdma_rpc.AllocPiece();
    }
    void free_small_piece(struct rpc* self, void* piece)
    {
        self->rdma_rpc.FreePiece(piece);
    }

    void rpc_handle_complete(struct rpc* self, struct rpc_execution* ex)
    {
        self->rdma_rpc.RpcServiceComplete((struct RpcExecution* )ex);

    }
    struct msg_base* get_request_header(struct rpc_execution* ex)
    {
        return (struct msg_base*)ex->req_msg;
    }

    struct msg_base* get_response_header(struct rpc_execution* ex)
    {
        return (struct msg_base*)ex->resp_msg;
    }

    int get_error_code(struct rpc_execution* ex)
    {
        return ex->cb_error;

    }


    struct hash_table
    {
        std::unordered_map<uint64_t, void*> table;
    };
    struct hash_table* create_hash_table()
    {
        return new struct hash_table;
    }

    void* hash_find(struct hash_table* t, uint64_t key)
    {
        std::unordered_map<uint64_t, void*>::iterator it = t->table.find(key);
        if (it != t->table.end()) {
            return it->second;
        }
        else {
            return NULL;
        }

    }

    void hash_insert(struct hash_table* t, uint64_t key, void* value)
    {
        t->table.insert({key, value});
    }

    void hash_delete(struct hash_table* t, uint64_t key)
    {
        t->table.erase(key);
    }

    void destroy_hash_table(struct hash_table* t)
    {
        delete t;
    }

    uint64_t hash_get(struct hash_table* t, uint64_t index)
    {
        std::unordered_map<uint64_t, void*>::iterator it;
        uint64_t i = 0;
        for (it = t->table.begin(); i<index&&it != t->table.end(); it++) {
            i++;
        }
        return it->first;
    }

    uint64_t hash_size(struct hash_table* t)
    {
        return t->table.size();
    }
    

    struct mini_queue {
        std::list<void*> mqueue;
    };

    struct mini_queue* create_queue() {
        return new struct mini_queue;
    }

    void destroy_queue(struct mini_queue* q) {
        delete q;
    }

    void enqueue(struct mini_queue* q, void* e) {
        q->mqueue.push_back(e);
    }

    void dequeue(struct mini_queue* q) {
        q->mqueue.pop_front();
    }

    void* queue_front(struct mini_queue* q) {
        return q->mqueue.front();
    }
    void queue_remove(struct mini_queue* q, void* e) {
        q->mqueue.remove(e);
     }
    int queue_size(struct mini_queue* q) {
        return q->mqueue.size();
    }

    double to_seconds(uint64_t cycles, double cycles_per_sec)
    {
        return Cycles::to_seconds(cycles, cycles_per_sec);

    }
    uint64_t to_microseconds(uint64_t cycles, double cycles_per_sec)
    {
        return Cycles::to_microseconds(cycles, cycles_per_sec);

    }
    uint64_t rdtsc()
    {
        return Cycles::rdtsc();
    }
#ifdef __cplusplus
};
#endif